//
// Copyright (C) 2024 Kumo group inc.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <krpc/rpc/controller.h>
#include <krpc/rpc/restful_request.h>
#include <krpc/rpc/restful_response.h>
#include <krpc/kthread/fiber.h>
#include <krpc/rpc/channel.h>
#include <turbo/strings/uri.h>
#include <turbo/utility/status.h>
#include <memory>
#include <future>
#include <turbo/times/time.h>

namespace krpc {

    class OnFutureCallback;
    class OnRestfulCallback;
    class RestfulClient;

    class RestfulTask {
    public:
        RestfulTask():_status(), _cntl(std::make_shared<Controller>()) {}

        ~RestfulTask() = default;

        RestfulRequest request(const std::string & url) {
            _raw_url = url;
            RestfulRequest req(_cntl.get());
            req.set_uri(url);
            return req;
        }

        void set_url(const std::string & url) {
            _raw_url = url;
            RestfulRequest req(_cntl.get());
            req.set_uri(url);
        }

        RestfulResponse response() {
            return RestfulResponse(_cntl.get());
        }

        bool ok() const {
            return _status.ok();
        }

        turbo::Status status() const {
            return _status;
        }

        void set_status(const turbo::Status &s) {
            _status = s;
        }

        Controller* controller() {
            return _cntl.get();
        }

        const std::string &url() const {
            return _raw_url;
        }

        turbo::Status wait_for(int64_t us) {
            if(!_status.ok()) {
                return _status;
            }
            DCHECK(event)<<"event no init";
            auto r = event->timed_wait(us);
            if( r != 0) {
                return turbo::errno_to_status(r, "");
            }
            return turbo::OkStatus();
        }

        turbo::Status wait_for(turbo::Duration d) {
            return wait_for(turbo::Duration::to_microseconds(d));
        }

        turbo::Status wait_until(turbo::Time t) {
            auto d = t - turbo::Time::current_time();
            if(d < turbo::Duration::zero()) {
                return turbo::deadline_exceeded_error("already timeout");
            }
            return wait_for(d);
        }

        turbo::Status wait() {
            if(!_status.ok()) {
                return _status;
            }
            DCHECK(event)<<"event no init";
            event->wait();
            return turbo::OkStatus();
        }

        void signal() {
            DCHECK(event)<<"event no init";
            event->decrease_signal();
        }
    private:
        void make_event() {
            DCHECK(!event)<<"already init event";
            event = std::make_shared<kthread::FiberCond>(1);
        }
    private:
        friend class RestfulClient;
        std::shared_ptr<kthread::FiberCond>   event;
        turbo::Status _status;
        std::shared_ptr<Controller> _cntl;
        std::string                 _raw_url;
    };

    turbo::Result<kutil::EndPoint> parse_endpoint(const turbo::Uri &uri);

    turbo::Result<kutil::EndPoint> parse_endpoint(const URI &uri);

    turbo::Result<kutil::EndPoint> parse_endpoint(const std::string &uri);

    class RestfulClient {
    public:

        static ChannelOptions get_default_options();
    public:
        RestfulClient(int64_t con = 100) :_concurrent_limit(con) {}

        RestfulTask get(const std::string &url);

        void get(RestfulTask *task);

        void async_get(const std::string &url,
                            std::function<void(const std::shared_ptr<RestfulTask> & task)> callback);

        void async_get(const std::shared_ptr<RestfulTask> & task,
                       std::function<void(const std::shared_ptr<RestfulTask> & task)> callback);

        void async_get(const std::shared_ptr<RestfulTask> & task);

        std::shared_ptr<RestfulTask> async_get(const std::string &url);

        int64_t concurrent() const {
            return _concurrent.load();
        }
    private:
        friend  class OnFutureCallback;
        friend class OnRestfulCallback;

        int64_t decrement() {
            return _concurrent.fetch_sub(1, std::memory_order_relaxed);
        }

        int64_t increment() {
            return _concurrent.fetch_add(1, std::memory_order_relaxed);
        }

    private:
        std::atomic<int64_t> _concurrent{0};
        int64_t              _concurrent_limit;
    };
}  // namespace krpc
